package com.rjhch.bigDataCluterHelp.hbaseHelp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * 表中region在所有regionServer中平衡分布
 * 要求：
 * 1. 每个regionServer上的region数量近似相同
 * 2. 每个表的region在所有regionServer上尽可能分散
 * 3. 尽可能少的迁移region
 */
public class TableBalance {
    private static final Logger log = LoggerFactory.getLogger(TableBalance.class);
    /**
     * 需要处理的regionEntities
     */
    private List<RegionEntity> regionEntities;
    /**
     * 系统中的regionServerNames
     */
    private String[] regionServerNames;
    /**
     * 系统中的表名列表（自动获取）
     */
    private List<String> tableNames = new ArrayList<>();
    /**
     * 每个表对应的region总数map（自动获取）
     */
    private Map<String, Integer> regionNumberOfTableMap = new HashMap<>();
    /**
     * 每个表对应的region在集群中的平均值map（自动获取）
     */
    private Map<String, Float> regionNumberAvgOfTableMap = new HashMap<>();
    /**
     * 所有region在集群中的平均值（自动获取）
     */
    private float allRegionNumberAvg;

    public TableBalance(List<RegionEntity> regionEntities, String[] regionServerNames) {
        this.regionEntities = regionEntities;
        log.info("regionEntities size: {}", regionEntities.size());
        log.debug("regionEntities: {}", regionEntities);
        this.regionServerNames = regionServerNames;
        log.info("regionServerNames: {}", Arrays.asList(regionServerNames));
        init();
    }

    /**
     * 初始化一些固定参数
     */
    private void init() {
        //init allRegionNumberAvg
        allRegionNumberAvg = (float) regionEntities.size() / (float) regionServerNames.length;
        log.info("allRegionNumberAvg: {}", allRegionNumberAvg);

        //init tableNames
        for (RegionEntity regionEntity : regionEntities) {
            if (!tableNames.contains(regionEntity.getTableName())) {
                tableNames.add(regionEntity.getTableName());
            }
        }
        log.info("tableNames: {}", tableNames);

        //init regionNumberOfTableMap
        for (String tableName : tableNames) {
            int number = 0;
            for (RegionEntity regionEntity : regionEntities) {
                if (regionEntity.getTableName().equals(tableName)) {
                    number++;
                }
            }
            regionNumberOfTableMap.put(tableName, number);
        }
        log.info("regionNumberOfTableMap: {}", regionNumberOfTableMap);

        //init regionNumberAvgOfTableMap
        for (String tableName : tableNames) {
            float regionNumberAvg = (float) regionNumberOfTableMap.get(tableName) / (float) regionServerNames.length;
            regionNumberAvgOfTableMap.put(tableName, regionNumberAvg);
        }
        log.info("regionNumberAvgOfTableMap: {}", regionNumberAvgOfTableMap);
    }

    /**
     * 进行平衡操作
     *
     * @return 返回需要迁移的region列表
     * @throws Exception
     */
    public List<RegionEntity> balance() throws Exception {
        log.info("start balance, regionEntities size: {} ...", regionEntities.size());

        //进行表级别的平衡
        for (String tableName : tableNames) {
            balanceTable(tableName);
        }

        //进行regionServer级别的平衡
        balanceRegionServer();

        List<RegionEntity> movedRegions = new ArrayList<>();
        for (RegionEntity regionEntity : regionEntities) {
            log.debug("result regionEntity: {} ", regionEntity);
            if (regionEntity.getMove()) {
                movedRegions.add(regionEntity);
            }
        }
        log.info("movedRegions size: {} ...", movedRegions.size());
        return movedRegions;
    }


    /**
     * 迁移表中region
     */
    private void balanceTable(String tableName) throws Exception {
        log.trace("regionEntities: {}", regionEntities);
        //获取每个regionServer上指定表的region总数map
        Map<String, Integer> regionNumberInRegionServerOfTableMap = getRegionNumberInRegionServerOfTable(tableName);
        //获取regionServer中region数量最大值时的regionServerName
        String regionServerNameOfMaxRegionNumber = getRegionServerNameOfMaxRegionNumber(regionNumberInRegionServerOfTableMap);
        log.debug("regionServer: {}, MaxRegionNumber: {}", regionServerNameOfMaxRegionNumber, regionNumberInRegionServerOfTableMap.get(regionServerNameOfMaxRegionNumber));
        //获取regionServer中region数量最小值时的regionServerName
        String regionServerNameOfMinRegionNumber = getRegionServerNameOfMinRegionNumber(regionNumberInRegionServerOfTableMap);
        int minRegionNumber = regionNumberInRegionServerOfTableMap.get(regionServerNameOfMinRegionNumber) == null ? 0 : regionNumberInRegionServerOfTableMap.get(regionServerNameOfMinRegionNumber);
        log.debug("regionServer: {}, MinRegionNumber: {}", regionServerNameOfMinRegionNumber, minRegionNumber);
        //只相差一个时不进行处理
        if (regionNumberInRegionServerOfTableMap.get(regionServerNameOfMaxRegionNumber) - minRegionNumber <= 1) {
            log.debug("table({}) will not balance when MaxRegionNumber and MinRegionNumber The difference is not greater than 1.", tableName);
            return;
        }
        //从最大值上的regionServer上迁移一个这个表的一个region到最小值的regionServer上
        RegionEntity regionEntity = getRegionEntityWithRegionServerNameAndTable(regionServerNameOfMaxRegionNumber, tableName);
        log.debug("regionEntity: {}", regionEntity);
        log.debug("move region {} from {} to {}", regionEntity.getRegionName(), regionServerNameOfMaxRegionNumber, regionServerNameOfMinRegionNumber);
        regionEntity.setMove(true);
        regionEntity.setRegionToServerName(regionServerNameOfMinRegionNumber);

        //重新启动本表平衡
        balanceTable(tableName);
    }

    /**
     * 平衡RegionServer
     */
    private void balanceRegionServer() throws Exception {
        log.trace("regionEntities: {}", regionEntities);
        //获取每个regionServer上指定表的region总数map
        Map<String, Integer> regionNumberInRegionServerMap = getRegionNumberMap();
        //获取regionServer中region数量最大值时的regionServerName
        String regionServerNameOfMaxRegionNumber = getRegionServerNameOfMaxRegionNumber(regionNumberInRegionServerMap);
        log.debug("regionServer: {}, MaxRegionNumber: {}", regionServerNameOfMaxRegionNumber, regionNumberInRegionServerMap.get(regionServerNameOfMaxRegionNumber));
        //获取regionServer中region数量最小值时的regionServerName
        String regionServerNameOfMinRegionNumber = getRegionServerNameOfMinRegionNumber(regionNumberInRegionServerMap);
        int minRegionNumber = regionNumberInRegionServerMap.get(regionServerNameOfMinRegionNumber) == null ? 0 : regionNumberInRegionServerMap.get(regionServerNameOfMinRegionNumber);
        log.debug("regionServer: {}, MinRegionNumber: {}", regionServerNameOfMinRegionNumber, minRegionNumber);
        //只相差一个时不进行处理
        if (regionNumberInRegionServerMap.get(regionServerNameOfMaxRegionNumber) - minRegionNumber <= 1) {
            log.debug("RegionServer will not balance when MaxRegionNumber and MinRegionNumber The difference is not greater than 1.");
            return;
        }
        //从最大值上的regionServer上迁移一个这个表的一个region到最小值的regionServer上
        RegionEntity regionEntity = getRegionEntityWithRegionServerName(regionServerNameOfMaxRegionNumber, regionServerNameOfMinRegionNumber);
        log.debug("regionEntity: {}", regionEntity);
        log.debug("move region {} from {} to {}", regionEntity.getRegionName(), regionServerNameOfMaxRegionNumber, regionServerNameOfMinRegionNumber);
        regionEntity.setMove(true);
        regionEntity.setRegionToServerName(regionServerNameOfMinRegionNumber);

        //重新启动regionServer的平衡
        balanceRegionServer();
    }

    /**
     * 获取每个regionServer上指定表的region总数
     */
    public Map<String, Integer> getRegionNumberInRegionServerOfTable(String tableName) {
        Map<String, Integer> numberMap = new HashMap<>();
        for (RegionEntity regionEntity : regionEntities) {
            if (!regionEntity.getTableName().equals(tableName)) {//不是指定表的情况
                continue;
            }

            String regionServerName;
            if (regionEntity.getMove()) {
                regionServerName = regionEntity.getRegionToServerName();
            } else {
                regionServerName = regionEntity.getRegionServerName();
            }

            if (numberMap.get(regionServerName) == null) {
                numberMap.put(regionServerName, 1);
            } else {
                numberMap.put(regionServerName, numberMap.get(regionServerName) + 1);
            }
        }
        log.trace("table: {}, region number in regionServer: {}", tableName, numberMap);
        return numberMap;
    }

    /**
     * 获取regionServer中region数量最大值时的regionServerName
     */
    private String getRegionServerNameOfMaxRegionNumber(Map<String, Integer> regionNumberInRegionServerOfTableMap) {
        int tmpRegionNumber = Integer.MIN_VALUE;
        String tmpRegionServerName = "";
        for (Map.Entry<String, Integer> entry : regionNumberInRegionServerOfTableMap.entrySet()) {
            if (entry.getValue() > tmpRegionNumber) {
                tmpRegionNumber = entry.getValue();
                tmpRegionServerName = entry.getKey();
            }
        }
        return tmpRegionServerName;
    }

    /**
     * 获取regionServer中一个表的region
     */
    private RegionEntity getRegionEntityWithRegionServerNameAndTable(String regionServerName, String tableName) throws Exception {
        List<RegionEntity> regionEntities = getRegionEntitiesWithRegionServerNameAndTable(regionServerName, tableName, false);
        if (regionEntities.isEmpty()) {
            throw new Exception("can not found regionEntity with regionServerName " + regionServerName + " and tableName " + tableName);
        }

        return regionEntities.get(0);//直接返回第一个即可
    }

    /**
     * 根据最大regionServerName和最小regionServerName获取要迁移的region
     */
    private RegionEntity getRegionEntityWithRegionServerName(String regionServerNameOfMaxRegionNumber, String regionServerNameOfMinRegionNumber) throws Exception {
        log.debug("regionServerNameOfMaxRegionNumber: {}, regionServerNameOfMinRegionNumber: {}", regionServerNameOfMaxRegionNumber, regionServerNameOfMinRegionNumber);
        for (String tableName : tableNames) {
            Map<String, Integer> regionNumberInRegionServerOfTable = getRegionNumberInRegionServerOfTable(tableName);
            if (regionNumberInRegionServerOfTable.get(regionServerNameOfMaxRegionNumber) == null) {//最大region数的regionServer没有这个表的region，检测下一张表
                continue;
            }
            if (regionNumberInRegionServerOfTable.get(regionServerNameOfMinRegionNumber) == null) {//如果在最小region数的regionServer没有这个表的region，则获取一个进行迁移
                log.debug("table: {}, min region number in regionServer {} has not this table region.", tableName, regionServerNameOfMinRegionNumber);
                return getRegionEntityWithRegionServerNameAndTable(regionServerNameOfMaxRegionNumber, tableName);
            }
            //如果在最大region数的regionServer上这个表的region数比在最小region数的regionServer上的大，则获取一个进行迁移
            if (regionNumberInRegionServerOfTable.get(regionServerNameOfMaxRegionNumber) > regionNumberInRegionServerOfTable.get(regionServerNameOfMinRegionNumber)) {
                log.debug("table: {}, max region number in regionServer is large then min region number in regionServer.", tableName);
                return getRegionEntityWithRegionServerNameAndTable(regionServerNameOfMaxRegionNumber, tableName);
            }
        }

        throw new Exception("can not found regionEntity with regionServerNameOfMaxRegionNumber " + regionServerNameOfMaxRegionNumber + " and regionServerNameOfMinRegionNumber " + regionServerNameOfMinRegionNumber);
    }

    /**
     * 获取regionServer中一个表的region
     */
    private List<RegionEntity> getRegionEntitiesWithRegionServerNameAndTable(String regionServerName, String tableName, boolean containMoved) {
        List<RegionEntity> tmpRegionEntities = new ArrayList<>();
        for (RegionEntity regionEntity : regionEntities) {
            if (!regionEntity.getTableName().equals(tableName)) {//不是指定表的情况
                continue;
            }
            if (!regionServerName.equals(regionEntity.getRegionServerName()) || regionServerName.equals(regionEntity.getRegionToServerName())) {//不是指定regionServer的情况
                continue;
            }
            if (!containMoved && regionEntity.getMove()) {//region已经迁移过
                continue;
            }
            tmpRegionEntities.add(regionEntity);
        }
        return tmpRegionEntities;
    }

    /**
     * 获取regionServer中region数量最小值时的regionServerName
     */
    private String getRegionServerNameOfMinRegionNumber(Map<String, Integer> regionNumberInRegionServerOfTableMap) {
        for (String regionServerName : regionServerNames) {
            if (regionNumberInRegionServerOfTableMap.get(regionServerName) == null) {//在此regionServer上没有找到此表对应的region
                return regionServerName;
            }
        }

        int tmpRegionNumber = Integer.MAX_VALUE;
        String tmpRegionServerName = "";
        for (Map.Entry<String, Integer> entry : regionNumberInRegionServerOfTableMap.entrySet()) {
            if (entry.getValue() < tmpRegionNumber) {
                tmpRegionNumber = entry.getValue();
                tmpRegionServerName = entry.getKey();
            }
        }
        return tmpRegionServerName;
    }

    /**
     * 获取每个regionServer上的region数量
     */
    public Map<String, Integer> getRegionNumberMap() {
        Map<String, Integer> numberMap = new HashMap<>();
        for (RegionEntity regionEntity : regionEntities) {
            String regionServerName;
            if (regionEntity.getMove()) {
                regionServerName = regionEntity.getRegionToServerName();
            } else {
                regionServerName = regionEntity.getRegionServerName();
            }

            if (numberMap.get(regionServerName) == null) {
                numberMap.put(regionServerName, 1);
            } else {
                numberMap.put(regionServerName, numberMap.get(regionServerName) + 1);
            }
        }
        log.trace("region number in regionServer: {}", numberMap);
        return numberMap;
    }


    public List<String> getTableNames() {
        return tableNames;
    }

    public Map<String, Float> getRegionNumberAvgOfTableMap() {
        return regionNumberAvgOfTableMap;
    }

    public float getAllRegionNumberAvg() {
        return allRegionNumberAvg;
    }
}
